package com.yanggu.spark.stream

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

//RDD队列
//测试过程中，可以通过使用ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列中的RDD，都会作为一个DStream处理。
object SparkStreaming02_QueueStream {

  def main(args: Array[String]): Unit = {

    //1. 创建sparkConf对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming02_QueueStream")

    //2. 创建StreamingContext
    val context = new StreamingContext(sparkConf, Seconds(4))

    //3. 创建RDD队列
    val rddQueue = new mutable.Queue[RDD[Int]]()

    //4. 创建QueueInputDStream
    val inputStream = context.queueStream(rddQueue, oneAtATime = false)

    //5. 处理队列中的数据
    val value = inputStream.map((_, 1)).reduceByKey(_ + _)

    //6. 打印结果
    value.print

    //7. 启动任务
    context.start

    //8. 循环创建RDD, 并向RDD队列中放入RDD
    for (_ <- 1 to 100) {
      rddQueue += context.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }

    context.awaitTermination
  }

}
